package com.platform.apex.cma.tools;

import java.io.File;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.List;
import java.util.UUID;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.datatorrent.api.DefaultOutputPort;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.platform.apex.cma.db.Grib2DataInfo;
import com.platform.apex.cma.db.Grib2DataInfoManager;
import com.platform.apex.util.FlieNameAndContent;
import com.platform.apex.util.HadoopFileOperUtil;
import com.platform.apex.util.KafkaMessageBus;

import ucar.ma2.Array;
import ucar.nc2.Dimension;
import ucar.nc2.NetcdfFile;
import ucar.nc2.Variable;

public class Launcher {

	private  HadoopFileOperUtil hdfsFileUtil;
	
	private  final Logger log = LoggerFactory.getLogger(Launcher.class);
	private  final ObjectMapper oMapper = ObjectMapperUtil.abtainObjectMapper();

	public void initHdfsTool(HadoopFileOperUtil hdfsFileUtil) {
		this.hdfsFileUtil = hdfsFileUtil;
	}
	
	public  void loadNCAndGenerateFile(List<String> paths,NetcdfFile ncfile,NetcdfFile ncfile1) {
		try {
			SimpleDateFormat sf=new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
			Grib2DataInfo grib2DataInfo = new Grib2DataInfo();
			Pattern pattern = Pattern.compile("[^0-9]");
//			NetcdfFile ncfile = null;
//			NetcdfFile ncfile1 = null;
			int year = 0,month = 0,day = 0,hour=0;
			Calendar calendar = Calendar.getInstance();
			if(paths.size() == 1){
				//ncfile = NetcdfDataset.open(input + "/" + paths.get(0));
				Variable v = ncfile.findVariable("hgt");
				List<Dimension> ldim = ncfile.getDimensions();
				int lon = 0, lat = 0, time = 0, level = 0;
				for(Dimension dim : ldim){
					String dims = dim.toString();
					Matcher matcher = pattern.matcher(dims);
					if(dims.indexOf("time")>-1)
						time = Integer.parseInt(matcher.replaceAll(""));
					else if(dims.indexOf("lon")>-1)
						lon = Integer.parseInt(matcher.replaceAll(""));
					else if(dims.indexOf("lat")>-1)
						lat = Integer.parseInt(matcher.replaceAll(""));
					else if(dims.indexOf("level")>-1)
						level = Integer.parseInt(matcher.replaceAll(""));
				}
				int[] levels = null;
				if(level == 17){
					levels = new int[]{2,5,9};
				}
				String str = "[{\"data\":[%s],\"header\":{\"surface1TypeName\":\"Isobaric surface\",\"discipline\":0,\"shapeName\":\"Earth spherical with radius of 6,371,229.0 m\",\"gridUnits\":\"degrees\","
						+ "\"numberPoints\":%s,\"resolution\":48,"
						+ "\"lo2\":357.5,\"lo1\":0,\"dx\":2.5,\"dy\":2.5,\"surface2Type\":255,\"productTypeName\":\"Forecast products\",\"productDefinitionTemplateName\":\"Analysis/forecast at horizontal level/layer at a point in time\",\"basicAngle\":0,\"disciplineName\":\"Meteorological products\",\"parameterNumberName\":\"Geopotential_height\",\"refTime\":\"2016-01-01T06:00:00.000Z\",\"surface1Type\":100,\"shape\":6,\"scanMode\":0,\"significanceOfRTName\":\"Start of forecast\",\"surface1Value\":50000.0,\"parameterUnit\":\"gpm\",\"productStatusName\":\"Operational products\",\"gridDefinitionTemplate\":0,\"gribLength\":70840,\"genProcessTypeName\":\"Forecast\",\"centerName\":\"US National Weather Service - NCEP(WMC)\",\"gridDefinitionTemplateName\":\"Latitude_Longitude\","
						+ "\"nx\":%s,\"parameterCategoryName\":\"Mass\",\"ny\":%s,\"surface2TypeName\":\"Missing\",\"subDivisions\":0,\"winds\":\"true\",\"forecastTime\":0,"
						+ "\"la2\":-90,\"la1\":90,\"productType\":1,\"parameterCategory\":3,\"center\":7,\"genProcessType\":2,\"subcenter\":0,\"productStatus\":0,\"productDefinitionTemplate\":0,\"parameterNumber\":5,\"surface2Value\":0.0,\"gribEdition\":2,\"significanceOfRT\":1}}]";
				
				year = Integer.valueOf(pattern.matcher(paths.get(0)).replaceAll(""));
				String folder="";
				Date dateStart = sf.parse(year + "-1-1 00:00:00");
				calendar.setTime(dateStart);
				
				for(int i = 0; i < time; i++){
					if(i!=0)
						calendar.add(Calendar.HOUR_OF_DAY, 6);
					for(int l : levels){
						Array values = v.read(new int[]{i,l,0,0},new int[]{1,1,lat,lon});
						hour = i%4;
						month = calendar.get(Calendar.MONTH)+1;
						day = calendar.get(Calendar.DATE);
						if(month==0)
							System.out.println(month);
						if(l == 2)
							folder = "Mass_Geopotential_height_Isobaric_surface_850";
						else if(l==5)
							folder = "Mass_Geopotential_height_Isobaric_surface_500";
						else if(l == 9)
							folder = "Mass_Geopotential_height_Isobaric_surface_200";
//						hdfsFileUtil.createFile(
//								"/eason_cma_grib2/cma/wrf_fnl/y_" + year + "/m_" + month + "/d_" + day + "/" + folder + "/h_" + (hour*6<10?"0"+hour*6:hour*6) + ".json",
//								String.format(str, values.toString().replace(" ", ",").substring(0,values.toString().length()-1),values.getSize(),lon,lat));
////						
						String filePath = "/eason_cma_grib2/cma/wrf_fnl/y_" + year + "/m_" + month + "/d_" + day + "/" + folder + "/h_" + (hour*6<10?"0"+hour*6:hour*6) + ".json";
						String content = String.format(str, values.toString().replace(" ", ",").substring(0,values.toString().length()-1),values.getSize(),lon,lat);
						KafkaMessageBus.put(new FlieNameAndContent(filePath, content));
					//	outhgt.emit(new FlieNameAndContent(filePath, content));
						long currentMills = System.currentTimeMillis();
						grib2DataInfo.setId(UUID.randomUUID().toString().replace("-", ""));
						grib2DataInfo.setCreateTime(currentMills);
						grib2DataInfo.setUpdateTime(currentMills);
//						grib2DataInfo.setAttributeName(folder);
						grib2DataInfo.setDateDayStr(year + "-" + (month < 10?"0"+month:month) + "-" + (day < 10?"0" + day : day) + " 00:00:00");
						Date date = sf.parse(year + "-" + (month < 10?"0" + month : month) + "-" + (day < 10?"0"+day : day) + " 00:00:00");
						grib2DataInfo.setDateDayLong(date.getTime());
						grib2DataInfo.setDateHourStr(year + "-" + (month < 10?"0" + month : month) + "-" + (day < 10?"0"+day : day) + " " + (hour*6 < 10?"0" + hour*6 : hour*6) + ":00:00");
						Date dateHour = sf.parse(year + "-" + (month < 10?"0" + month : month) + "-" + (day < 10?"0"+day : day) + " " + (hour*6 < 10?"0" + hour*6 : hour*6) + ":00:00");  
						grib2DataInfo.setDateHourLong(dateHour.getTime());
						grib2DataInfo.setFilePath("/eason_cma_grib2/cma/wrf_fnl/y_" + year + "/m_" + month + "/d_" + day + "/%s/h_" + (hour*6 < 10?"0" + hour*6 : hour*6) + ".json");
						Grib2DataInfoManager.create(grib2DataInfo);
						ncfile=null;
					}
				}
			}
			else{
//				ncfile = NetcdfDataset.open(input + "/" + paths.get(0));
//				ncfile1 = NetcdfDataset.open(input + "/" + paths.get(1));
				
				Variable u = ncfile.findVariable("uwnd");
				Variable v = ncfile1.findVariable("vwnd");
				List<Dimension> ldim = ncfile.getDimensions();
				int lon = 0, lat = 0, time = 0, level = 0;
				for(Dimension dim : ldim){
					String dims = dim.toString();
					Matcher matcher = pattern.matcher(dims);
					if(dims.indexOf("time")>-1)
						time = Integer.parseInt(matcher.replaceAll(""));
					else if(dims.indexOf("lon")>-1)
						lon = Integer.parseInt(matcher.replaceAll(""));
					else if(dims.indexOf("lat")>-1)
						lat = Integer.parseInt(matcher.replaceAll(""));
					else if(dims.indexOf("level")>-1)
						level = Integer.parseInt(matcher.replaceAll(""));
				}
				int[] levels = null;
				if(level == 17){
					levels = new int[]{2,5,9,16};
				}
				String str = "[{\"data\":[%s],\"header\":{\"surface1TypeName\":\"Specified height level above ground\",\"discipline\":0,\"shapeName\":\"Earth spherical with radius of 6,371,229.0 m\",\"gridUnits\":\"degrees\","
						+ "\"numberPoints\":%s,\"resolution\":48,\"lo2\":357.5,\"lo1\":0,\"dx\":2.5,\"dy\":2.5,\"surface2Type\":255,\"productTypeName\":\"Forecast products\",\"productDefinitionTemplateName\":\"Analysis/forecast at horizontal level/layer at a point in time\",\"basicAngle\":0,\"disciplineName\":\"Meteorological products\",\"parameterNumberName\":\"U-component_of_wind\",\"refTime\":\"2016-01-01T12:00:00.000Z\",\"surface1Type\":103,\"shape\":6,\"scanMode\":0,\"significanceOfRTName\":\"Start of forecast\",\"surface1Value\":10.0,\"parameterUnit\":\"m.s-1\",\"productStatusName\":\"Operational products\",\"gridDefinitionTemplate\":0,\"gribLength\":76311,\"genProcessTypeName\":\"Forecast\",\"centerName\":\"US National Weather Service - NCEP(WMC)\",\"gridDefinitionTemplateName\":\"Latitude_Longitude\",\"nx\":%s,\"parameterCategoryName\":\"Momentum\",\"ny\":%s,\"surface2TypeName\":\"Missing\",\"subDivisions\":0,\"winds\":\"true\",\"forecastTime\":0,\"la2\":-90,\"la1\":90,\"productType\":1,\"parameterCategory\":2,\"center\":7,\"genProcessType\":2,\"subcenter\":0,\"productStatus\":0,\"productDefinitionTemplate\":0,\"parameterNumber\":2,\"surface2Value\":0.0,\"gribEdition\":2,\"significanceOfRT\":1}},"
						+ "{\"data\":[%s],\"header\":{\"surface1TypeName\":\"Specified height level above ground\",\"discipline\":0,\"shapeName\":\"Earth spherical with radius of 6,371,229.0 m\",\"gridUnits\":\"degrees\","
						+ "\"numberPoints\":%s,\"resolution\":48,\"lo2\":357.5,\"lo1\":0,\"dx\":2.5,\"dy\":2.5,\"surface2Type\":255,\"productTypeName\":\"Forecast products\",\"productDefinitionTemplateName\":\"Analysis/forecast at horizontal level/layer at a point in time\",\"basicAngle\":0,\"disciplineName\":\"Meteorological products\",\"parameterNumberName\":\"V-component_of_wind\",\"refTime\":\"2016-01-01T12:00:00.000Z\",\"surface1Type\":103,\"shape\":6,\"scanMode\":0,\"significanceOfRTName\":\"Start of forecast\",\"surface1Value\":10.0,\"parameterUnit\":\"m.s-1\",\"productStatusName\":\"Operational products\",\"gridDefinitionTemplate\":0,\"gribLength\":76215,\"genProcessTypeName\":\"Forecast\",\"centerName\":\"US National Weather Service - NCEP(WMC)\",\"gridDefinitionTemplateName\":\"Latitude_Longitude\",\"nx\":%s,\"parameterCategoryName\":\"Momentum\",\"ny\":%s,\"surface2TypeName\":\"Missing\",\"subDivisions\":0,\"winds\":\"true\",\"forecastTime\":0,\"la2\":-90,\"la1\":90,\"productType\":1,\"parameterCategory\":2,\"center\":7,\"genProcessType\":2,\"subcenter\":0,\"productStatus\":0,\"productDefinitionTemplate\":0,\"parameterNumber\":3,\"surface2Value\":0.0,\"gribEdition\":2,\"significanceOfRT\":1}}]";
				
				year = Integer.valueOf(pattern.matcher(paths.get(0)).replaceAll(""));
				Date dateStart = sf.parse(year + "-01-01 00:00:00");
				calendar.setTime(dateStart);
				String folder = "Momentum_Component_of_wind_Specified_height_level_above_ground_10";
				for(int i = 0; i < time; i++){
					if(i!=0)
						calendar.add(Calendar.HOUR_OF_DAY, 6);
					for(int l : levels){
						if(l==16)
							folder = "Momentum_Component_of_wind_Isobaric_surface_850";
						else if(l == 9)
							folder = "Momentum_Component_of_wind_Isobaric_surface_500";
						else if(l == 5)
							folder = "Momentum_Component_of_wind_Isobaric_surface_200";
						else
							folder = "Momentum_Component_of_wind_Isobaric_surface_10";
						Array valueu = u.read(new int[]{i,l,0,0},new int[]{1,1,lat,lon});
						Array valuev = v.read(new int[]{i,l,0,0},new int[]{1,1,lat,lon});
						hour = i%4;
						month = calendar.get(Calendar.MONTH)+1;
						day = calendar.get(Calendar.DATE);
//						hdfsFileUtil.createFile(
//								"/eason_cma_grib2/cma/wrf_fnl/y_" + year + "/m_" + month + "/d_" + day + "/" + folder + "/h_" + (hour*6<10?"0"+hour*6:hour*6) + ".json",
//								String.format(str, valueu.toString().replace(" ", ",").substring(0,valueu.toString().length()-1), valueu.getSize(), lon, lat, valuev.toString().replace(" ", ",").substring(0,valuev.toString().length()-1), valueu.getSize(), lon, lat));
//	
						String filePath = "/eason_cma_grib2/cma/wrf_fnl/y_" + year + "/m_" + month + "/d_" + day + "/" + folder + "/h_" + (hour*6<10?"0"+hour*6:hour*6) + ".json";
						String content = String.format(str, valueu.toString().replace(" ", ",").substring(0,valueu.toString().length()-1), valueu.getSize(), lon, lat, valuev.toString().replace(" ", ",").substring(0,valuev.toString().length()-1), valueu.getSize(), lon, lat);
						KafkaMessageBus.put(new FlieNameAndContent(filePath, content));
						//outhgt.emit(new FlieNameAndContent(filePath, content));
						
						long currentMills = System.currentTimeMillis();
						grib2DataInfo.setId(UUID.randomUUID().toString().replace("-", ""));
						grib2DataInfo.setCreateTime(currentMills);
						grib2DataInfo.setUpdateTime(currentMills);
						grib2DataInfo.setAttributeName(folder);
						grib2DataInfo.setDateDayStr(year + "-" + (month < 10?"0"+month:month) + "-" + (day < 10?"0" + day : day) + " 00:00:00");
						Date date = sf.parse(year + "-" + (month < 10?"0" + month : month) + "-" + (day < 10?"0"+day : day) + " 00:00:00");
						grib2DataInfo.setDateDayLong(date.getTime());
						grib2DataInfo.setDateHourStr(year + "-" + (month < 10?"0" + month : month) + "-" + (day < 10?"0"+day : day) + " " + (hour*6 < 10?"0" + hour*6 : hour*6) + ":00:00");
						Date dateHour = sf.parse(year + "-" + (month < 10?"0" + month : month) + "-" + (day < 10?"0"+day : day) + " " + (hour*6 < 10?"0" + hour*6 : hour*6) + ":00:00");  
						grib2DataInfo.setDateHourLong(dateHour.getTime());
						grib2DataInfo.setFilePath("/eason_cma_grib2/cma/wrf_fnl/y_" + year + "/m_" + month + "/d_" + day + "/%s/h_" + (hour*6 < 10?"0" + hour*6 : hour*6) + ".json");
						Grib2DataInfoManager.create(grib2DataInfo);
					}
				}
			}
		} catch (Exception e) {
			log.info("file '" + "', analysis faile, error info: ", e);
		}
	}


	public  void traverseFolder(String inputFilePath, String outputFilePath) {

		File file = new File(inputFilePath);
		if (file.exists()) {
			File[] files = file.listFiles();
			if (files.length > 0) {
				for (File realFile : files) {
					if (realFile.isDirectory()) {
						traverseFolder(realFile.getAbsolutePath(), outputFilePath);
					}else{
						List<String> paths = new ArrayList<>();
						if(realFile.getName().indexOf("hgt")>-1){
							paths.add(realFile.getName());
							loadNCAndGenerateFile(paths,null,null);
						}
						else if(realFile.getName().indexOf("uwnd")>-1){
							paths.add(realFile.getName());
							paths.add(realFile.getName().replace("uwnd", "vwnd"));
							loadNCAndGenerateFile(paths,null,null);
						}
					}
				}
			}
		} else {
			log.info("folder '" + inputFilePath + "', not exist.");
		}
	}

	
}
